package com.star.learn.excel.service;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.star.learn.excel.domain.Student;
import com.star.learn.excel.util.RedisConstants;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class RedisMetadataService {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    private final ObjectMapper objectMapper = new ObjectMapper(); // 用于序列化/反序列化实际数据

    // 获取元数据的 Redis Key
    private String getMetaKey(String fileId) {
        return RedisConstants.META_KEY_PREFIX + fileId;
    }

    // 获取实际批次数据的 Redis Key
    private String getDataKey(String fileId, String sheetName, int batchNum) {
        return String.format("%s%s:%s:%d", RedisConstants.EXCEL_DATA_KEY_PREFIX, fileId, sheetName, batchNum);
    }

    /**
     * 标记文件为处理中，设置开始时间，并设置较长的过期时间
     */
    public void markFileProcessing(String fileId) {
        String metaKey = getMetaKey(fileId);
        try {
            redisTemplate.opsForHash().put(metaKey, "status", "PROCESSING");
            redisTemplate.opsForHash().put(metaKey, "startTime", String.valueOf(System.currentTimeMillis()));
            // 设置较长的过期时间，例如 24小时
            redisTemplate.expire(metaKey, 24, TimeUnit.HOURS);
            log.info("文件 {} 标记为 PROCESSING 状态.", fileId);
        } catch (Exception e) {
            log.error("标记文件 {} 为 PROCESSING 失败.", fileId, e);
            throw new RuntimeException("标记文件状态失败", e); // 向上抛出，让调用者决定如何处理
        }
    }

    /**
     * 获取文件当前状态
     */
    public Object getFileStatus(String fileId) {
        String metaKey = getMetaKey(fileId);
        try {
            return redisTemplate.opsForHash().get(metaKey, "status");
        } catch (Exception e) {
            log.error("获取文件 {} 状态失败.", fileId, e);
            // 返回 null 或抛出异常
            return null;
        }
    }

    /**
     * 使用 Redis HINCRBY 实现原子批次号自增
     * @param fileId 组成 Redis 哈希键 metaKey，例如：meta:fileId
     * @param sheetName 当前 sheet 名称
     * @return 当前批次号（从 1 开始）
     */
    public int getNextBatchNumber(String fileId, String sheetName) {
        String metaKey = getMetaKey(fileId);
        String batchIndexKey = sheetName + "_batchIndex";
        try {
            // 原子自增（不存在时自动初始化为 0，再加 1）
            Long nextBatch = redisTemplate.opsForHash().increment(metaKey, batchIndexKey, 1);
            return nextBatch.intValue();
        } catch (Exception e) {
            log.error("获取文件 {} Sheet {} 的下一批次号失败.", fileId, sheetName, e);
            throw new RuntimeException("获取批次号失败", e);
        }
    }

    /**
     * 更新批次元数据状态 (PENDING 或 COMPLETED)
     */
    public void updateBatchMetadata(String fileId, String sheetName, int batchNum, int batchSize, String status) {
        String metaKey = getMetaKey(fileId);
        String batchKey = String.format("%s:%d", sheetName, batchNum);
        try {
            Map<String, String> batchMeta = getBatchMetadata(fileId, sheetName, batchNum);
            if(batchMeta == null) {
                batchMeta = new HashMap<>();
            }
            // 更新或设置字段
            batchMeta.put("count", String.valueOf(batchSize));
            batchMeta.put("time", LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
            batchMeta.put("status", status);
            // 存储批次元数据
            redisTemplate.opsForHash().put(metaKey, batchKey, batchMeta);
            log.debug("文件 {} Sheet {} Batch {}: 元数据状态更新为 {}", fileId, sheetName, batchNum, status);

            // 再次设置较长的过期时间，防止元数据在处理过程中过期
            redisTemplate.expire(metaKey, 24, TimeUnit.HOURS);

        } catch (Exception e) {
            log.error("文件 {} Sheet {} Batch {}: 更新元数据状态 {} 或触发时间失败.", fileId, sheetName, batchNum, status, e);
            // 注意：这里失败不抛出异常，以免影响消费者 ACK/NACK 决策
        }
    }

    /**
     * 将批次数据存储到 Redis
     */
    public void saveBatchData(String fileId, String sheetName, int batchNum, List<Student> dataList) {
        String dataKey = getDataKey(fileId, sheetName, batchNum);
        try {
            ObjectMapper mapper = new ObjectMapper();
            String jsonData = mapper.writeValueAsString(dataList);
            // 将批次数据存储为 JSON 字符串
            redisTemplate.opsForValue().set(dataKey, jsonData);
            // 设置批次数据的过期时间，例如 1 小时
            redisTemplate.expire(dataKey, 1, TimeUnit.HOURS);
            log.debug("文件 {} Sheet {} Batch {}: 批次数据已存储到 Redis.", fileId, sheetName, batchNum);
        } catch (Exception e) {
            log.error("文件 {} Sheet {} Batch {}: 存储批次数据到 Redis 失败.", fileId, sheetName, batchNum, e);
            throw new RuntimeException("存储批次数据到 Redis 失败", e);
        }
    }

    /**
     * 从 Redis 获取批次数据
     */
    public List<Student> getBatchData(String fileId, String sheetName, int batchNum) {
        String dataKey = getDataKey(fileId, sheetName, batchNum);
        try {
            String jsonData = (String) redisTemplate.opsForValue().get(dataKey);
            if (jsonData == null) {
                return null; // 数据可能已过期或被删除
            }
            ObjectMapper mapper = new ObjectMapper();
            return mapper.readValue(jsonData, new TypeReference<>() {});
        } catch (Exception e) {
            log.error("文件 {} Sheet {} Batch {}: 从 Redis 获取批次数据失败.", fileId, sheetName, batchNum, e);
            // 返回 null 或空列表，让调用者判断如何处理
            return null;
        }
    }

    /**
     * 新增：从 Redis 获取单个批次的元数据
     * @return 批次元数据 Map，如果不存在则返回 null
     */
    public Map<String, String> getBatchMetadata(String fileId, String sheetName, int batchNum) {
        String metaKey = getMetaKey(fileId);
        String batchKey = String.format("%s:%d", sheetName, batchNum);
        try {
            Object batchMetaObj = redisTemplate.opsForHash().get(metaKey, batchKey);
            if (batchMetaObj instanceof Map<?, ?> rawMap) {
                // 由于 RedisTemplate 默认序列化可能导致 value 是 LinkedHashMap<Object, Object>
                // 需要进行类型转换，或者配置更合适的序列化器
                Map<String, String> stringMap = new HashMap<>();
                rawMap.forEach((k, v) -> stringMap.put(k.toString(), v != null ? v.toString() : null));
                return stringMap;
            } else if (batchMetaObj != null) {
                log.warn("文件 {} Sheet {} Batch {}: Redis 元数据格式非 Map, 实际类型: {}",
                        fileId, sheetName, batchNum, batchMetaObj.getClass().getName());
            }
            return null;
        } catch (Exception e) {
            log.error("文件 {} Sheet {} Batch {}: 获取批次元数据失败.", fileId, sheetName, batchNum, e);
            return null;
        }
    }

    /**
     * 从 Redis 删除批次数据 (消费者成功处理后调用)
     */
    public void deleteBatchData(String fileId, String sheetName, int batchNum) {
        String dataKey = getDataKey(fileId, sheetName, batchNum);
        try {
            redisTemplate.delete(dataKey);
            log.debug("文件 {} Sheet {} Batch {}: 批次数据已从 Redis 删除.", fileId, sheetName, batchNum);
        } catch (Exception e) {
            log.error("文件 {} Sheet {} Batch {}: 从 Redis 删除批次数据失败.", fileId, sheetName, batchNum, e);
            // 不抛出异常，日志记录即可
        }
    }

    /**
     * 在消费者成功入库后，原子地累加总记录数
     */
    public void incrementTotalCount(String fileId, int count) {
        String metaKey = getMetaKey(fileId);
        try {
            redisTemplate.opsForHash().increment(metaKey, "total", count);
            log.debug("文件 {}: 总记录数累加 {}", fileId, count);
        } catch (Exception e) {
            log.error("文件 {}: 累加总记录数 {} 失败.", fileId, count, e);
            // 不抛出异常
        }
    }

    /**
     * 检查文件所有批次是否完成，并标记整个文件为 COMPLETED
     */
    public void checkAndMarkFileComplete(String fileId) {
        String metaKey = getMetaKey(fileId);
        try {
            Map<Object, Object> entries = redisTemplate.opsForHash().entries(metaKey);

            // 检查文件整体状态是否已经是 COMPLETED，避免重复检查
            Object fileStatus = entries.get("status");
            if ("COMPLETED".equals(fileStatus)) {
                log.debug("文件 {} 已是 COMPLETED 状态，跳过检查.", fileId);
                return;
            }

            boolean allCompleted = entries.entrySet().stream()
                    // 过滤出表示批次元数据的字段 (键中包含 ':' 但不是以 '_batchIndex' 结尾)
                    .filter(e -> {
                        String keyStr = e.getKey().toString();
                        return keyStr.contains(":") && !keyStr.endsWith("_batchIndex");
                    })
                    .allMatch(e -> {
                        Object val = e.getValue();
                        if (val instanceof Map<?, ?> rawMap) {
                            // 需要确保这里能正确处理 Map<Object, Object> 或 Map<String, String>
                            Object statusObj = rawMap.get("status");
                            return "COMPLETED".equals(statusObj != null ? statusObj.toString() : null);
                        }
                        log.warn("文件 {}: 元数据格式异常，键: {}, 值: {}", fileId, e.getKey(), val);
                        return false; // 格式异常的批次不认为是完成的
                    });

            if (allCompleted) {
                redisTemplate.opsForHash().put(metaKey, "status", "COMPLETED");
                redisTemplate.opsForHash().put(metaKey, "endTime", String.valueOf(System.currentTimeMillis()));

                Object startObj = redisTemplate.opsForHash().get(metaKey, "startTime");
                Object totalObj = redisTemplate.opsForHash().get(metaKey, "total");
                String totalCount = totalObj != null ? totalObj.toString() : "N/A";

                if (startObj != null) {
                    long duration = System.currentTimeMillis() - Long.parseLong(startObj.toString());
                    log.info("文件 {} 全链路处理完成. 总记录数: {}. 总耗时: {} ms", fileId, totalCount, duration);
                } else {
                    log.info("文件 {} 全链路处理完成. 总记录数: {}. 开始时间未知.", fileId, totalCount);
                }
            } else {
                log.debug("文件 {}: 仍有批次未完成，暂不标记文件为 COMPLETED.", fileId);
            }
        } catch (Exception e) {
            log.error("文件 {}: 检查或标记文件完成状态时发生 Redis 错误.", fileId, e);
        }
    }

    /**
     * 查找长时间处于 PENDING 状态的批次
     * 使用 SCAN 命令安全地迭代 Key
     * @return 符合条件的批次列表 (fileId, sheetName, batchNum, batchSize 的 Map 列表)
     */
    public List<Map<String, Object>> findStuckPendingBatches() {
        List<Map<String, Object>> stuckBatches = new ArrayList<>();

        try {
            // 使用 SCAN 命令迭代 meta keys
            ScanOptions scanOptions = ScanOptions.scanOptions().match(RedisConstants.META_KEY_PREFIX + "*").count(100).build();
            Cursor<byte[]> cursor = null;
            try {
                // 获取低级别连接以执行 SCAN
                cursor = redisTemplate.getConnectionFactory().getConnection().scan(scanOptions);
                while (cursor.hasNext()) {
                    String metaKey = new String(cursor.next());
                    String fileId = metaKey.substring(RedisConstants.META_KEY_PREFIX.length());

                    try {
                        // 获取该 meta key (Hash 类型) 下的所有字段和值
                        Map<Object, Object> entries = redisTemplate.opsForHash().entries(metaKey);
                        if (entries.isEmpty()) {
                            continue;
                        }

                        // 遍历 Hash 中的每个 entry (字段-值对)
                        entries.forEach((key, value) -> {
                            String batchKey = key.toString();
                            // 过滤出表示批次元数据的字段 (例如 "Sheet1:1")
                            if (batchKey.contains(":") && !batchKey.endsWith("_batchIndex")) {
                                if (value instanceof Map) {
                                    Map<String, String> batchMeta = (Map<String, String>) value;
                                    String status = batchMeta.get("status");

                                    // 仅检查状态是否为 PENDING
                                    if ("PENDING".equals(status)) {
                                        String[] parts = batchKey.split(":");
                                        if (parts.length == 2) {
                                            try {
                                                String sheetName = parts[0];
                                                int batchNum = Integer.parseInt(parts[1]);
                                                int batchSize = Integer.parseInt(
                                                        batchMeta.getOrDefault("count", "0")
                                                );

                                                Map<String, Object> batchInfo = new HashMap<>();
                                                batchInfo.put("fileId", fileId);
                                                batchInfo.put("sheetName", sheetName);
                                                batchInfo.put("batchNum", batchNum);
                                                batchInfo.put("batchSize", batchSize);
                                                stuckBatches.add(batchInfo);
                                                log.debug("发现 PENDING 批次: {}", batchKey);
                                            } catch (NumberFormatException e) {
                                                log.warn("批次 Key 格式异常: {}", batchKey);
                                            }
                                        }
                                    }
                                } else {
                                    log.warn("文件 {}: 忽略非 Map 类型的批次元数据值，键: {}", fileId, batchKey);
                                }
                            }
                        });
                    } catch (Exception e) {
                        log.error("扫描文件 {} 的元数据时发生错误.", fileId, e);
                        // 继续扫描下一个 meta key
                    }
                }
            } finally {
                if (cursor != null) {
                    try {
                        cursor.close();
                    } catch (Exception e) {
                        log.error("关闭 Redis SCAN 游标失败", e);
                    }
                }
            }

        } catch (Exception e) {
            log.error("扫描 Redis Key 查找 stuck PENDING 批次时发生错误.", e);
        }

        return stuckBatches;
    }

    /**
     * 设置元数据在 Redis 中的过期时间 (例如在 BackupRecoveryTask 中恢复后调用)
     */
    public void setMetaExpire(String fileId, long timeout, TimeUnit unit) {
        String metaKey = getMetaKey(fileId);
        try {
            redisTemplate.expire(metaKey, timeout, unit);
        } catch (Exception e) {
            log.error("设置文件 {} 元数据过期时间失败.", fileId, e);
        }
    }
}